package Logistic

import breeze.numerics.abs
import org.apache.spark.sql.functions.{avg, col, round, when}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.ml.classification.{LogisticRegression, LogisticRegressionModel}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{VectorAssembler, VectorIndexer}
import org.apache.spark.sql.types.DoubleType

object Logistic {
  def main(args: Array[String]): Unit = {
    // 实例化对象
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("score")
      .getOrCreate()

    // 设置日志等级
    spark.sparkContext.setLogLevel("WARN")

    // 保留列名
    var Dataframe: DataFrame = spark.read.option("header", true).csv("E:\\code\\java\\SparkAPI\\src\\main\\resources\\smalltest.csv")

    // 删除血糖为Null的数据（没有判断的数据就没有意义）
    val arr: Array[String] = Array("血糖", "id")
    Dataframe = Dataframe.na.drop(arr)

    // 狗键Null填充Map(map有效、其他无效)
    // Map=====     Null默认转化为  0      ======
    // 因为正常值的概率比不正常的概率大，保守使用正常值（这里就不查正常值了）。
    var map = Map(("乙肝e抗体", 0), ("乙肝e抗原", 0), ("乙肝核心抗体", 0), ("乙肝表面抗体", 0), ("乙肝表面抗原", 0))
    for (i <- Dataframe.columns) {
      map = map + (i -> 0) // 添加元素
    }

    println("Null值数量" + Dataframe.na.drop().count())
    // 更新Dataframe(Null 已经 全部转化为 0)
    Dataframe = Dataframe.na.fill(map)

    //    Dataframe.withColumn("label",when(col("血糖")>7,1)
    //      .otherwise(0))
    //      .groupBy(col("label"))
    //      .agg(count("id"),avg("红细胞平均血红蛋白量"),avg("红细胞计数")).show()

    //血糖已经没有用了。被Label列代替了
    val DataFrameAfterDecide = Dataframe.withColumn("label", when(col("血糖") > 7, 1)
      .otherwise(0)).drop(col("血糖"))

    //更新到Dataframe，后面训练模型需要用(存有label)
    Dataframe = DataFrameAfterDecide

    val DatasetAfterGroupBy = DataFrameAfterDecide.groupBy(col("label"))

    DatasetAfterGroupBy.agg(avg("红细胞计数")).show()

    val a =indexDateFrame(1,1,DatasetAfterGroupBy.agg(avg("红细胞计数")))
    println(a)

    println(DatasetAfterGroupBy.agg(avg("红细胞计数")).head().get(1))

    // 查看全部属性的差值
    val mapSubResult: Map[String, Double] = avgAndSubtractAnswer(DataFrameAfterDecide)
    println(mapSubResult)

    // 选出差距较大的属性，进行Logistics回归
    val Answer: Map[String, Double] = mapSubResult.filter(x => abs(x._2) > 3)
    val deleteName: Map[String, Double] = mapSubResult.filter(x => abs(x._2) < 3)
    println("Answer: "+Answer)
    println("deleteName: "+deleteName)

    /*=======Binomial logistic regression======*/

    var DataframeTest: DataFrame = spark.read.option("header", true).csv("E:\\code\\java\\SparkAPI\\src\\main\\resources\\test.csv")


    var maptest = Map(("乙肝e抗体", 0))
    for (i <- DataframeTest.columns) {
      maptest = maptest + (i -> 0) // 添加元素
    }
    DataframeTest = DataframeTest.na.fill(maptest)

    // 训练集和预测集都删去多余属性
    for (col_name<-deleteName.keys)
      {
        // 避免把“label"删了
        if (col_name!="label")
        Dataframe=Dataframe.drop(col(col_name))
        DataframeTest=DataframeTest.drop(col(col_name))
      }

    //将所有列的String转化为数值
    for (coltamp<-Dataframe.columns)
      {
        Dataframe = Dataframe.withColumn(coltamp,col(coltamp).cast(DoubleType))
      }

    for (coltamp<-DataframeTest.columns)
    {
      DataframeTest = DataframeTest.withColumn(coltamp,col(coltamp).cast(DoubleType))
    }

    // 多列转换 ==  label为列名了
    val Train = new VectorAssembler()
      .setInputCols(Answer.keys.toArray)
      .setOutputCol("features")
      .transform(Dataframe)

    // 拆分训练集，分成训练部分和测试部分
    val Array(train,test): Array[Dataset[Row]] = Train.randomSplit(Array(0.75, 0.25))

    val DFpredict = new VectorAssembler()
      .setInputCols(Answer.keys.toArray)
      .setOutputCol("features")
      .transform(DataframeTest)

    // 获取模型
    val Model: LogisticRegressionModel = logisticGetModel(train)

    // 对训练集进行测试评估  prediction不能被评估器识别
    val predict = Model.transform(test).withColumn("predict",col("prediction"))


    // 获取分类问题的评估器
    val evaluator = getClassificationEvaluator()

    // 根据准确率对模型进行评估
    val accuracy =evaluator.evaluate(predict)

    // 结果如下：accuracy: 0.9272727272727272
    println("accuracy: "+accuracy)

    // 对待判定集合进行预测
    val AfterPredictData: DataFrame = logisticToFit(test, Model)

    // 稍微查看一下
    AfterPredictData.show(10)

    /* 小数据集训练的模型成功执行。将文件名改成源数据集train即可*/
  }

  /*
  * Author:luojian
  * E-mail:984344367@qq
  * Time:2021年11月18日14:14:36
  * Dateset or DataFrame index
  * @prams row:行数
  *        col:列数
  *        dataFrame: 用于索引的DataFrame或DataSet
  * @Return: Long
  * */
  def indexDateFrame(row:Int,col:Int,dataFrame: DataFrame): Double ={
    var i = 0
    for (rowTemp <-dataFrame.head(row))
    {
      // 获取第一列不能用getLong，要用getInt
      if (i==row-1 && col!=1 && rowTemp.get(col-1)!=null){
        return  rowTemp.get(col-1).toString.trim.toDouble // 可以考虑一下getDouble试试
      }
      else if (i==row-1 && col==1) {
        return rowTemp.getInt(0).toString.trim.toDouble
      }
      i=i+1
    }
    return -1
  }

/*方法二：使用SQL的连接逻辑实现分组均值的差值处理
      val fun1 = udf{(x:Int,y:Int)=>x-y}
      var AvgList =  List()
        for (i<-Dataframe.columns)
          {
            var temp = DatasetAfterGroupBy.agg(avg(col(i)))
            temp.crossJoin(temp).agg(fun1(col(""),col("")))
            AvgList=AvgList+
          }
      }
      function1 分组均值求差
          def function1(a: Int, b: Int): Int = {
            return a - b
          }
 */

  /*
  * Author:luojian
  * E-mail:984344367@qq
  * Time:2021年11月18日14:45:25
  * */
  def avgAndSubtractAnswer(dataFrame: DataFrame): Map[String,Double] ={
    var mapTemp = Map[String,Double]()
    var temp :Double = 0; // In Subtract
    val AfterGroupBy = dataFrame.groupBy(col("label"))
    for (colname<-dataFrame.columns)
      {
        val colTemp=AfterGroupBy.agg(round(avg(col(colname)),2))
        temp = indexDateFrame(1,2,colTemp)-indexDateFrame(2,2,colTemp)
        mapTemp=mapTemp+(colname->temp)
      }
    return mapTemp
  }

  def logisticGetModel(dataFrame: DataFrame): LogisticRegressionModel ={
    val lr = new LogisticRegression()
      .setMaxIter(10)
      .setRegParam(0.3)
      .setElasticNetParam(0.8)
    val lrModel = lr.fit(dataFrame)
    return lrModel
  }


  def logisticToFit(test:DataFrame,model: LogisticRegressionModel): DataFrame ={
    val frame: DataFrame = model.transform(test)
    return frame
  }


  def getClassificationEvaluator(): MulticlassClassificationEvaluator ={
    val evaluator = new MulticlassClassificationEvaluator()
      .setLabelCol("label")
      .setPredictionCol("predict")
      .setMetricName("accuracy")
    return evaluator
  }
}
